// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <array>
#include <atomic>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <utility>
#include <vector>

#include <pollux/common/file/file.h>
#include <pollux/common/file/region.h>
#include <pollux/common/io/io_statistics.h>
#include <pollux/dwio/common/metrics_log.h>

namespace kumo::pollux::dwio::common {
    using namespace kumo::pollux::io;

    /**
     * An abstract interface for providing readers a stream of bytes.
     */
    class InputStream {
    public:
        explicit InputStream(
            const std::string &path,
            const MetricsLogPtr &metricsLog = MetricsLog::voidLog(),
            IoStatistics *stats = nullptr,
            filesystems::File::IoStats *fsStats = nullptr)
            : path_{path},
              metricsLog_{metricsLog},
              stats_(stats),
              fsStats_(fsStats) {
        }

        virtual ~InputStream() = default;

        /**
         * Get the stats object
         */
        IoStatistics *getStats() const {
            return stats_;
        }

        /**
         * Get the total length of the file in bytes.
         */
        virtual uint64_t getLength() const = 0;

        /**
         * Get the natural size for reads.
         * @return the number of bytes that should be read at once
         */
        virtual uint64_t getNaturalReadSize() const = 0;

        /**
         * Read length bytes from the file starting at offset into
         * the buffer starting at buf.
         * @param buf the starting position of a buffer.
         * @param length the number of bytes to read.
         * @param offset the position in the stream to read from.
         */
        virtual void read(void *, uint64_t, uint64_t, LogType) = 0;

        /**
         * Read starting at offset into buffers, filling the buffers left to right. A
         * buffer with data() == nullptr indicates a gap in the read, so that its
         * size() worth bytes are skipped. There must be enough data in 'this' to
         * cover the sum of the sizes of 'buffers'.
         *
         * @buffers - The buffers to read into.
         * @param offset the position in the stream to read from.
         */
        virtual void read(
            const std::vector<melon::Range<char *> > &buffers,
            uint64_t offset,
            LogType logType) {
            uint64_t bufferOffset = 0;
            for (auto &range: buffers) {
                if (range.data()) {
                    read(range.data(), range.size(), offset + bufferOffset, logType);
                }
                bufferOffset += range.size();
            }
        }

        /// Like read() with the same arguments but returns the result or
        /// exception via SemiFuture. Use only if hasReadAsync() is true.
        virtual melon::SemiFuture<uint64_t> readAsync(
            const std::vector<melon::Range<char *> > &buffers,
            uint64_t offset,
            LogType logType);

        /// Returns true if readAsync has a native implementation that is
        /// asynchronous.
        virtual bool hasReadAsync() const {
            return false;
        }

        /**
         * Take advantage of vectorized read API provided by some file system.
         * Allow file system to do optimzied reading plan to disk to minimize
         * total bytes transferred through network. Stores the result in an IOBuf
         * range named at `iobufs`, which must have the same size as `regions`.
         */
        virtual void vread(
            melon::Range<const pollux::common::Region *> regions,
            melon::Range<melon::IOBuf *> iobufs,
            const LogType purpose) = 0;

        const std::string &getName() const;

        virtual void logRead(uint64_t offset, uint64_t length, LogType purpose);

    protected:
        std::string path_;
        MetricsLogPtr metricsLog_;
        IoStatistics *stats_;
        filesystems::File::IoStats *fsStats_;
    };

    /// An input stream that reads from an already opened ReadFile.
    class ReadFileInputStream final : public InputStream {
    public:
        /// Takes shared ownership of |readFile|.
        explicit ReadFileInputStream(
            std::shared_ptr<pollux::ReadFile>,
            const MetricsLogPtr &metricsLog = MetricsLog::voidLog(),
            IoStatistics *stats = nullptr,
            filesystems::File::IoStats *fsStats = nullptr);

        ~ReadFileInputStream() override = default;

        uint64_t getLength() const final override {
            return readFile_->size();
        }

        uint64_t getNaturalReadSize() const final override {
            return readFile_->getNaturalReadSize();
        }

        void read(void *, uint64_t, uint64_t, LogType) override;

        void read(
            const std::vector<melon::Range<char *> > &buffers,
            uint64_t offset,
            LogType logType) override;

        melon::SemiFuture<uint64_t> readAsync(
            const std::vector<melon::Range<char *> > &buffers,
            uint64_t offset,
            LogType logType) override;

        bool hasReadAsync() const override;

        void vread(
            melon::Range<const pollux::common::Region *> regions,
            melon::Range<melon::IOBuf *> iobufs,
            const LogType purpose) override;

        const std::shared_ptr<pollux::ReadFile> &getReadFile() const {
            return readFile_;
        }

    private:
        std::shared_ptr<pollux::ReadFile> readFile_;
    };
} // namespace kumo::pollux::dwio::common
